#include "gtest/gtest.h"

#include <chrono>
#include <atomic>
#include <filesystem>
#include <fstream>
#include <string>
#include <stdexcept>
#include <memory>
#include <thread>
#include <vector>

#include "asycipc/api.hpp"
#include "asycipc/backpressure.hpp"
#include "asycipc/config_service.hpp"
#include "asycipc/crc32c.hpp"
#include "asycipc/frame.hpp"
#include "asycipc/handshake.hpp"
#include "asycipc/heartbeat.hpp"
#include "asycipc/inmemory_transport.hpp"
#include "asycipc/server_runtime.hpp"
#include "asycipc/session_context.hpp"
#include "asycipc/sliding_window.hpp"
#include "asycipc/telemetry.hpp"
#ifdef _WIN32
#include "asycipc/named_pipe.hpp"
#endif

namespace asycipc {
namespace {

TEST(Crc32cTest, MatchesReferenceVector) {
  const std::vector<uint8_t> data = {'1', '2', '3', '4', '5', '6', '7', '8', '9'};
  EXPECT_EQ(Crc32c(data.data(), data.size()), 0xE3069283u);
}

TEST(FrameTest, RoundTripPreservesFields) {
  std::vector<uint8_t> payload = {'p', 'i', 'n', 'g'};
  std::vector<uint8_t> extensions = {0x01, 0x02, 0x03};
  Frame frame = Frame::MakeData(42, payload, extensions, /*flags=*/0x05);

  const auto raw = frame.Serialize();
  Frame parsed = Frame::Deserialize(raw);

  EXPECT_EQ(parsed.magic, kFrameMagic);
  EXPECT_EQ(parsed.version, kProtocolVersion);
  EXPECT_EQ(parsed.type, FrameType::kData);
  EXPECT_EQ(parsed.seq_no, 42u);
  EXPECT_EQ(parsed.flags, 0x05);
  EXPECT_EQ(parsed.payload, payload);
  EXPECT_EQ(parsed.extensions, extensions);
}

TEST(SessionContextTest, EnforcesStateTransitions) {
  SessionContext session;
  EXPECT_EQ(session.state(), SessionState::kCreated);

  session.TransitionTo(SessionState::kNegotiating);
  session.TransitionTo(SessionState::kActive);

  const auto seq = session.NextSendSeq();
  EXPECT_EQ(seq, 1u);
  session.AckReceived(seq);
  session.RegisterInbound(1);

  EXPECT_THROW(session.TransitionTo(SessionState::kCreated),
               SessionStateError);
  session.TransitionTo(SessionState::kDraining);
  session.TransitionTo(SessionState::kClosed);
}

TEST(HandshakeManagerTest, NegotiatesWhenAppAllowed) {
  HandshakeConfig config;
  config.endpoint_name = "asycipc";
  config.allowed_app_ids.insert("demo-app");
  HandshakeManager manager(config);

  SessionContext session;
  HandshakeRequest request;
  request.app_id = "demo-app";

  HandshakeResponse response = manager.Negotiate(request, session);

  EXPECT_EQ(session.state(), SessionState::kActive);
  EXPECT_FALSE(response.session_id.empty());
  EXPECT_EQ(response.initial_seq, 1u);
  EXPECT_FALSE(response.resume_token.empty());
}

TEST(SlidingWindowTest, RetriesAndAckWork) {
  SlidingWindow window(/*window_size=*/2, std::chrono::milliseconds(100), 3);
  const auto now = std::chrono::steady_clock::now();
  window.Push(Frame::MakeData(1, {'a'}), now);
  window.Push(Frame::MakeData(2, {'b'}), now);
  EXPECT_TRUE(window.IsFull());

  auto retries = window.CollectRetries(now + std::chrono::milliseconds(120));
  ASSERT_EQ(retries.size(), 2u);
  EXPECT_EQ(retries[0].frame.seq_no, 1u);
  EXPECT_EQ(retries[1].frame.seq_no, 2u);

  EXPECT_TRUE(window.MarkAck(1));
  EXPECT_EQ(window.PendingCount(), 1u);
  EXPECT_TRUE(window.MarkNack(2, now));

  retries = window.CollectRetries(now + std::chrono::milliseconds(10));
  ASSERT_EQ(retries.size(), 1u);
  EXPECT_EQ(retries[0].frame.seq_no, 2u);
}

TEST(HeartbeatMonitorTest, DetectsTimeoutAndSend) {
  const auto start = std::chrono::steady_clock::now();
  HeartbeatMonitor monitor(std::chrono::milliseconds(100), 1, start);
  EXPECT_FALSE(monitor.ShouldSend(start));
  EXPECT_FALSE(monitor.IsExpired(start));

  const auto send_time = start + std::chrono::milliseconds(120);
  EXPECT_TRUE(monitor.ShouldSend(send_time));
  monitor.MarkSent(send_time);
  EXPECT_FALSE(monitor.ShouldSend(send_time));

  monitor.MarkReceived(send_time);
  const auto timeout_time = send_time + std::chrono::milliseconds(250);
  EXPECT_TRUE(monitor.IsExpired(timeout_time));
}

TEST(BackpressureControllerTest, EmitsTransitions) {
  BackpressureController controller(/*high*/ 4, /*low*/ 2);

  std::vector<bool> events;
  controller.SetHandler([&](bool active) { events.push_back(active); });

  controller.Update(3);
  EXPECT_TRUE(events.empty());

  controller.Update(5);
  ASSERT_EQ(events.size(), 1u);
  EXPECT_TRUE(events.back());
  EXPECT_TRUE(controller.IsBackpressured());

  controller.Update(1);
  ASSERT_EQ(events.size(), 2u);
  EXPECT_FALSE(events.back());
  EXPECT_FALSE(controller.IsBackpressured());
}

TEST(ConfigServiceTest, LoadsAndReloadsConfig) {
  ConfigService service;
  const auto path = std::filesystem::temp_directory_path() / "asycipc_config_test.json";

  {
    std::ofstream out(path);
    out << R"({"retry":{"base_ms":200},"logging":{"level":"info"}})";
  }

  ASSERT_TRUE(service.LoadFromFile(path.string()));
  EXPECT_EQ(service.GetInt("retry.base_ms", 0), 200);
  EXPECT_EQ(service.GetString("logging.level", ""), "info");

  bool notified = false;
  service.RegisterListener("logging", [&]() { notified = true; });

  std::this_thread::sleep_for(std::chrono::milliseconds(5));
  {
    std::ofstream out(path);
    out << R"({"retry":{"base_ms":400},"logging":{"level":"debug"}})";
  }

  ASSERT_TRUE(service.ReloadIfChanged());
  EXPECT_TRUE(notified);
  EXPECT_EQ(service.GetInt("retry.base_ms", 0), 400);
  EXPECT_EQ(service.GetString("logging.level", ""), "debug");

  std::filesystem::remove(path);
}

TEST(TelemetryCenterTest, TracksCountersGaugesAndHistograms) {
  TelemetryCenter telemetry;

  telemetry.IncrementCounter("ipc_send_total");
  telemetry.IncrementCounter("ipc_send_total", 4);
  EXPECT_EQ(telemetry.GetCounter("ipc_send_total"), 5u);

  telemetry.SetGauge("ipc_latency_avg", 2.5);
  EXPECT_DOUBLE_EQ(telemetry.GetGauge("ipc_latency_avg"), 2.5);

  telemetry.ObserveHistogram("ipc_latency", 5.0);
  telemetry.ObserveHistogram("ipc_latency", 7.5);
  const auto values = telemetry.GetHistogram("ipc_latency");
  ASSERT_EQ(values.size(), 2u);
  EXPECT_DOUBLE_EQ(values[0], 5.0);
  EXPECT_DOUBLE_EQ(values[1], 7.5);
}

TEST(ChannelReliabilityTest, IntegratesModules) {
  SessionContext session;
  session.TransitionTo(SessionState::kNegotiating);
  session.TransitionTo(SessionState::kActive);
  Channel channel(session, nullptr);

  std::vector<bool> events;
  channel.SetBackpressureHandler([&](bool active) { events.push_back(active); });

  auto window = std::make_shared<SlidingWindow>(2, std::chrono::milliseconds(100), 2);
  auto now = std::chrono::steady_clock::now();
  auto heartbeat = std::make_shared<HeartbeatMonitor>(std::chrono::milliseconds(50), 1, now);
  auto controller = std::make_shared<BackpressureController>(2, 1);
  TelemetryCenter telemetry;
  channel.SetReliabilityModules(window, heartbeat, controller, &telemetry);

  std::vector<Frame> delivered;
  std::vector<std::string> errors;
  channel.SetMessageHandler([&](Channel& channel, const Frame& frame) {
    (void)channel;
    delivered.push_back(frame);
    return std::vector<Frame>{};
  });
  channel.SetErrorHandler([&](const std::string& msg) { errors.push_back(msg); });

  std::chrono::steady_clock::time_point current = now;
  channel.SetTimeProvider([&]() { return current; });

  auto f1_frames = channel.MakeDataFrames({'x'});
  ASSERT_EQ(f1_frames.size(), 1u);
  auto f1 = f1_frames.front();
  auto f2_frames = channel.MakeDataFrames({'y'});
  ASSERT_EQ(f2_frames.size(), 1u);
  auto f2 = f2_frames.front();

  ASSERT_EQ(f1.flags & kFrameFlagAckRequired, kFrameFlagAckRequired);
  ASSERT_EQ(f2.flags & kFrameFlagAckRequired, kFrameFlagAckRequired);
  ASSERT_EQ(events.size(), 1u);
  EXPECT_TRUE(events.back());
  EXPECT_EQ(telemetry.GetCounter("ipc_send_total"), 2u);

  auto ack1 = Frame::MakeAck(f1.seq_no);
  auto ack_result = channel.HandleInboundFrame(ack1);
  EXPECT_FALSE(ack_result.application_frame.has_value());
  EXPECT_TRUE(ack_result.control_frames.empty());
  ASSERT_EQ(events.size(), 2u);
  EXPECT_FALSE(events.back());
  EXPECT_EQ(window->PendingCount(), 1u);

  current += std::chrono::milliseconds(120);
  auto retries = channel.PollRetries();
  ASSERT_EQ(retries.size(), 1u);
  EXPECT_EQ(retries[0].frame.seq_no, f2.seq_no);
  EXPECT_EQ(telemetry.GetCounter("ipc_retry_total"), 1u);

  auto inbound_seq = channel.session().expected_recv_seq();
  Frame inbound = Frame::MakeData(inbound_seq, {'z'}, {}, kFrameFlagAckRequired);
  auto inbound_result = channel.HandleInboundFrame(inbound);
  EXPECT_FALSE(inbound_result.application_frame.has_value());
  ASSERT_EQ(inbound_result.control_frames.size(), 1u);
  EXPECT_EQ(inbound_result.control_frames[0].type, FrameType::kAck);
  EXPECT_EQ(inbound_result.control_frames[0].seq_no, inbound.seq_no);
  EXPECT_EQ(telemetry.GetCounter("ipc_recv_total"), 1u);
  ASSERT_EQ(delivered.size(), 1u);
  EXPECT_EQ(delivered.back().payload, (std::vector<uint8_t>{'z'}));

  current += std::chrono::milliseconds(60);
  EXPECT_TRUE(channel.ShouldSendHeartbeat());
  channel.MarkHeartbeatSent();
  EXPECT_EQ(telemetry.GetCounter("ipc_heartbeat_sent"), 1u);
  EXPECT_FALSE(channel.ShouldSendHeartbeat());

  auto heartbeat_ack = Frame::MakeHeartbeatAck();
  auto hb_ack_result = channel.HandleInboundFrame(heartbeat_ack);
  EXPECT_FALSE(hb_ack_result.application_frame.has_value());
  EXPECT_EQ(telemetry.GetCounter("ipc_heartbeat_recv"), 1u);

  auto heartbeat_frame = Frame::MakeHeartbeat();
  auto hb_result = channel.HandleInboundFrame(heartbeat_frame);
  ASSERT_EQ(hb_result.control_frames.size(), 1u);
  EXPECT_EQ(hb_result.control_frames[0].type, FrameType::kHeartbeatAck);

  current += std::chrono::milliseconds(120);
  EXPECT_TRUE(channel.HeartbeatExpired());

  events.clear();
  channel.HandleInboundFrame(Frame::MakeAck(f2.seq_no));
  EXPECT_EQ(window->PendingCount(), 0u);
  EXPECT_TRUE(events.empty());
  EXPECT_DOUBLE_EQ(telemetry.GetGauge("ipc_backpressure_state"), 0.0);
  EXPECT_EQ(telemetry.GetCounter("ipc_ack_total"), 2u);
  EXPECT_TRUE(errors.empty());
}

TEST(ChannelReliabilityTest, ReportsHandlerError) {
  SessionContext session;
  session.TransitionTo(SessionState::kNegotiating);
  session.TransitionTo(SessionState::kActive);
  Channel channel(session, nullptr);
  TelemetryCenter telemetry;
  channel.SetReliabilityModules(nullptr, nullptr, nullptr, &telemetry);

  std::vector<std::string> errors;
  channel.SetMessageHandler([&](Channel& channel, const Frame&) -> std::vector<Frame> {
    (void)channel;
    throw std::runtime_error("boom");
    return {};
  });
  channel.SetErrorHandler([&](const std::string& msg) { errors.push_back(msg); });

  auto seq = channel.session().expected_recv_seq();
  auto inbound = Frame::MakeData(seq, {'q'}, {}, kFrameFlagAckRequired);
  auto result = channel.HandleInboundFrame(inbound);

  ASSERT_TRUE(result.application_frame.has_value());
  EXPECT_EQ(result.application_frame->payload, (std::vector<uint8_t>{'q'}));
  ASSERT_EQ(result.control_frames.size(), 1u);
  EXPECT_EQ(result.control_frames[0].type, FrameType::kAck);
  EXPECT_EQ(errors.size(), 1u);
  EXPECT_EQ(errors.front(), "boom");
  EXPECT_EQ(telemetry.GetCounter("ipc_recv_total"), 1u);
}

TEST(ChannelFragmentationTest, AutoSplitsAndReassemblesPayload) {
  SessionContext send_session;
  send_session.TransitionTo(SessionState::kNegotiating);
  send_session.TransitionTo(SessionState::kActive);
  Channel sender(send_session, nullptr);

  const std::size_t large_size = kMaxFramePayloadBytes * 2 + 128;
  std::vector<uint8_t> large_payload(large_size);
  for (std::size_t i = 0; i < large_payload.size(); ++i) {
    large_payload[i] = static_cast<uint8_t>(i % 251);
  }

  auto outgoing_frames = sender.MakeDataFrames(large_payload);
  ASSERT_GT(outgoing_frames.size(), 1u);

  SessionContext recv_session;
  recv_session.TransitionTo(SessionState::kNegotiating);
  recv_session.TransitionTo(SessionState::kActive);
  Channel receiver(recv_session, nullptr);
  TelemetryCenter telemetry;
  receiver.SetReliabilityModules(nullptr, nullptr, nullptr, &telemetry);

  std::vector<Frame> delivered;
  receiver.SetMessageHandler([&](Channel&, const Frame& frame) {
    delivered.push_back(frame);
    return std::vector<Frame>{};
  });
  std::vector<std::string> errors;
  receiver.SetErrorHandler([&](const std::string& msg) { errors.push_back(msg); });

  for (const auto& frag : outgoing_frames) {
    auto inbound = receiver.HandleInboundFrame(frag);
    if (frag.flags & kFrameFlagAckRequired) {
      ASSERT_FALSE(inbound.control_frames.empty());
      EXPECT_EQ(inbound.control_frames.back().type, FrameType::kAck);
    }
    EXPECT_FALSE(inbound.application_frame.has_value());
  }

  ASSERT_EQ(delivered.size(), 1u);
  EXPECT_EQ(delivered.front().payload, large_payload);
  EXPECT_TRUE(errors.empty());
}

TEST(PipeServerTest, ChannelsBridgeHandlers) {
  std::vector<Frame> messages;
  std::vector<std::string> errors;
  auto server = PipeServer::Builder()
                     .Endpoint("asycipc")
                     .AllowApp("demo-app")
                    .OnMessage([&](Channel& channel, const Frame& frame) {
                      (void)channel;
                      messages.push_back(frame);
                      return std::vector<Frame>{};
                    })
                     .OnError([&](const std::string& msg) { errors.push_back(msg); })
                     .Build();
  server.Start();

  auto client = PipeClient::Builder().Endpoint("asycipc").AppId("demo-app").Build();
  auto request = client.BuildHandshakeRequest();

  SessionContext session;
  HandshakeResponse response = server.Accept(request, session);
  EXPECT_FALSE(response.session_id.empty());
  auto channel = server.CreateChannel(std::move(session));
  ASSERT_NE(channel, nullptr);

  auto seq = channel->session().expected_recv_seq();
  Frame inbound = Frame::MakeData(seq, {'p'}, {}, kFrameFlagAckRequired);
  auto control = server.ProcessInbound(*channel, inbound.Serialize());

  ASSERT_EQ(control.size(), 1u);
  EXPECT_EQ(control[0].type, FrameType::kAck);
  EXPECT_EQ(control[0].seq_no, inbound.seq_no);
  ASSERT_EQ(messages.size(), 1u);
  EXPECT_EQ(messages.back().payload, (std::vector<uint8_t>{'p'}));
  EXPECT_TRUE(errors.empty());
  EXPECT_EQ(server.telemetry().GetCounter("ipc_recv_total"), 1u);
}

TEST(PipeServerTest, MessageHandlerFragmentsLargeReply) {
  const std::size_t reply_size = kMaxFramePayloadBytes * 3 + 42;
  std::vector<uint8_t> reply_payload(reply_size, 0x7Au);

  auto server = PipeServer::Builder()
                     .Endpoint("asycipc")
                     .AllowApp("demo-app")
                     .OnMessage([&](Channel& channel, const Frame&) {
                       return channel.MakeDataFrames(reply_payload);
                     })
                     .Build();
  server.Start();

  auto client = PipeClient::Builder().Endpoint("asycipc").AppId("demo-app").Build();
  SessionContext session;
  server.Accept(client.BuildHandshakeRequest(), session);
  auto channel = server.CreateChannel(std::move(session));
  ASSERT_NE(channel, nullptr);

  auto inbound_seq = channel->session().expected_recv_seq();
  Frame inbound = Frame::MakeData(inbound_seq, {'p'}, {}, kFrameFlagAckRequired);

  std::vector<Frame> responses = server.ProcessInbound(*channel, inbound.Serialize());

  const std::size_t expected_chunks =
      (reply_payload.size() + kMaxFramePayloadBytes - 1) / kMaxFramePayloadBytes;
  ASSERT_EQ(responses.size(), expected_chunks + 1);  // ack + fragments
  EXPECT_EQ(responses.front().type, FrameType::kAck);

  SessionContext recv_session;
  recv_session.TransitionTo(SessionState::kNegotiating);
  recv_session.TransitionTo(SessionState::kActive);
  Channel receiver(recv_session, nullptr);
  std::vector<Frame> delivered;
  receiver.SetMessageHandler([&](Channel&, const Frame& frame) {
    delivered.push_back(frame);
    return std::vector<Frame>{};
  });

  for (std::size_t i = 1; i < responses.size(); ++i) {
    const auto& frag = responses[i];
    EXPECT_EQ(frag.type, FrameType::kData);
    EXPECT_TRUE((frag.flags & kFrameFlagFragment) != 0);
    auto res = receiver.HandleInboundFrame(frag);
    ASSERT_FALSE(res.control_frames.empty());
    EXPECT_EQ(res.control_frames.back().type, FrameType::kAck);
  }

  ASSERT_EQ(delivered.size(), 1u);
  EXPECT_EQ(delivered.front().payload, reply_payload);
}

TEST(PipeServerTest, ReportsInvalidFrameError) {
  std::vector<std::string> errors;
  auto server = PipeServer::Builder()
                     .Endpoint("asycipc")
                     .AllowApp("demo-app")
                     .OnError([&](const std::string& msg) { errors.push_back(msg); })
                     .Build();
  server.Start();
  auto client = PipeClient::Builder().Endpoint("asycipc").AppId("demo-app").Build();
  SessionContext session;
  server.Accept(client.BuildHandshakeRequest(), session);
  auto channel = server.CreateChannel(std::move(session));
  ASSERT_NE(channel, nullptr);

  std::vector<uint8_t> garbage = {0x00, 0x01, 0x02};
  auto control = server.ProcessInbound(*channel, garbage);
  EXPECT_TRUE(control.empty());
  ASSERT_FALSE(errors.empty());
}

TEST(PipeServerTest, CollectsMaintenanceFrames) {
  auto server = PipeServer::Builder().Endpoint("asycipc").AllowApp("demo-app").Build();
  server.Start();
  auto client = PipeClient::Builder().Endpoint("asycipc").AppId("demo-app").Build();
  SessionContext session;
  server.Accept(client.BuildHandshakeRequest(), session);
  auto channel = server.CreateChannel(std::move(session));
  ASSERT_NE(channel, nullptr);

  auto start = std::chrono::steady_clock::now();
  std::chrono::steady_clock::time_point now = start;
  channel->SetTimeProvider([&]() { return now; });

  auto outbound_frames = channel->MakeDataFrames({'d'});
  ASSERT_EQ(outbound_frames.size(), 1u);
  auto outbound = outbound_frames.front();
  now += std::chrono::milliseconds(6000);
  auto frames = server.CollectMaintenanceFrames(*channel);

  ASSERT_EQ(frames.size(), 2u);
  EXPECT_EQ(frames[0].type, FrameType::kData);
  EXPECT_EQ(frames[0].seq_no, outbound.seq_no);
  EXPECT_EQ(frames[1].type, FrameType::kHeartbeat);
}

TEST(ServerConnectionTest, PumpOnceSendsAckAndMaintenance) {
  InMemoryPipePair pair;
  auto server_handle = pair.ServerHandle();
  auto client_handle = pair.ClientHandle();

  std::vector<Frame> delivered;
  auto server = PipeServer::Builder()
                     .Endpoint("asycipc")
                     .AllowApp("demo-app")
                     .OnMessage([&](Channel& channel, const Frame& frame) {
                       delivered.push_back(frame);
                       return channel.MakeDataFrames({'r', 'e', 'p', 'l', 'y'});
                     })
                     .Build();
  server.Start();
  auto client = PipeClient::Builder().Endpoint("asycipc").AppId("demo-app").Build();

  SessionContext session;
  server.Accept(client.BuildHandshakeRequest(), session);
  auto channel = server.CreateChannel(std::move(session));
  ASSERT_NE(channel, nullptr);

  std::chrono::steady_clock::time_point now = std::chrono::steady_clock::now();
  channel->SetTimeProvider([&]() { return now; });

  ServerConnection connection(server, std::move(server_handle), std::move(channel));

  auto seq = connection.channel().session().expected_recv_seq();
  auto inbound = Frame::MakeData(seq, {'x'}, {}, kFrameFlagAckRequired);
  client_handle->Write(inbound.Serialize());

  ASSERT_TRUE(connection.PumpOnce());
  ASSERT_EQ(delivered.size(), 1u);
  EXPECT_EQ(delivered.back().payload, (std::vector<uint8_t>{'x'}));

  const auto ack_bytes = client_handle->Read();
  auto ack_frame = Frame::Deserialize(ack_bytes);
  EXPECT_EQ(ack_frame.type, FrameType::kAck);
  EXPECT_EQ(ack_frame.seq_no, inbound.seq_no);

  const auto response_bytes = client_handle->Read();
  auto response_frame = Frame::Deserialize(response_bytes);
  EXPECT_EQ(response_frame.type, FrameType::kData);
  EXPECT_EQ(response_frame.payload, (std::vector<uint8_t>{'r', 'e', 'p', 'l', 'y'}));

  connection.channel().HandleInboundFrame(Frame::MakeAck(response_frame.seq_no));

  now += std::chrono::milliseconds(6000);
  connection.FlushMaintenance();
  const auto hb_bytes = client_handle->Read();
  auto hb_frame = Frame::Deserialize(hb_bytes);
  EXPECT_EQ(hb_frame.type, FrameType::kHeartbeat);
}

TEST(InMemoryTransportTest, ExchangesPayloadsSynchronously) {
  InMemoryPipePair pair;
  auto server_handle = pair.ServerHandle();
  auto client_handle = pair.ClientHandle();

  const std::vector<uint8_t> payload = {'p', 'o', 'n', 'g'};
  client_handle->Write(payload);

  const auto received = server_handle->Read();
  EXPECT_EQ(received, payload);
}

#ifdef _WIN32
TEST(WindowsPipeTest, NamedPipeRoundTrip) {
  const auto token = static_cast<unsigned long long>(
      std::chrono::steady_clock::now().time_since_epoch().count());
  const std::string pipe_name =
      "\\\\.\\pipe\\asycipc_test_pipe_" + std::to_string(token);

  auto listener = CreateNamedPipeListener({pipe_name});
  auto* listener_ptr = listener.get();

  std::thread server([&]() {
    auto handle = listener_ptr->Accept();
    const auto data = handle->Read();
    EXPECT_EQ(data, (std::vector<uint8_t>{'p', 'i', 'n', 'g'}));
    handle->Write(std::vector<uint8_t>{'o', 'k'});
  });

  std::this_thread::sleep_for(std::chrono::milliseconds(50));

  auto client = ConnectNamedPipeEndpoint(
      {pipe_name, std::chrono::milliseconds(5000), std::chrono::milliseconds(20)});

  const std::vector<uint8_t> payload = {'p', 'i', 'n', 'g'};
  client->Write(payload);

  const auto buffer = client->Read();
  EXPECT_EQ(buffer, (std::vector<uint8_t>{'o', 'k'}));

  server.join();
}

TEST(WindowsPipeServeTest, TelemetryReflectsIocpLifecycle) {
  std::atomic<bool> message_seen{false};
  std::atomic<int> error_count{0};
  std::atomic<int> handshake_calls{0};

  auto server = PipeServer::Builder()
                     .Endpoint("telemetry")
                     .OnMessage([&](Channel& channel, const Frame&) {
                       (void)channel;
                       message_seen.store(true);
                       return std::vector<Frame>{};
                     })
                     .OnError([&](const std::string&) { error_count.fetch_add(1); })
                     .Build();
  server.Start();

  const auto token = static_cast<unsigned long long>(
      std::chrono::steady_clock::now().time_since_epoch().count());
  const std::string pipe_name =
      "\\\\.\\pipe\\asycipc_iocp_test_" + std::to_string(token);

  auto listener = CreateNamedPipeListener({pipe_name});
  auto* listener_ptr = listener.get();

  std::atomic<bool> stop{false};
  auto handshake = [&](PipeServer&, PipeHandle&, SessionContext& session) {
    const int call_index = handshake_calls.fetch_add(1);
    if (call_index > 0) {
      stop.store(true);
      return false;
    }
    session.SetSessionId("session");
    session.TransitionTo(SessionState::kNegotiating);
    session.TransitionTo(SessionState::kActive);
    return true;
  };

  std::thread server_thread([&]() {
    ServeNamedPipe(server, *listener_ptr, handshake, std::chrono::milliseconds(50),
                   [&stop]() { return !stop.load(); }, 2, 1024);
  });

  std::this_thread::sleep_for(std::chrono::milliseconds(50));

  auto connect_client = [&]() -> PipeHandlePtr {
    try {
      return ConnectNamedPipeEndpoint(
          {pipe_name, std::chrono::milliseconds(5000), std::chrono::milliseconds(20)});
    } catch (...) {
      return {};
    }
  };

  auto client = connect_client();
  ASSERT_TRUE(client) << "failed to connect first client";

  Frame data = Frame::MakeData(1, std::vector<uint8_t>{'p', 'i', 'n', 'g'}, {},
                               kFrameFlagAckRequired);
  auto serialized = data.Serialize();
  client->Write(serialized);

  const auto buffer = client->Read();
  EXPECT_FALSE(buffer.empty());
  EXPECT_TRUE(message_seen.load());

  Frame ack_frame;
  ASSERT_NO_THROW(ack_frame = Frame::Deserialize(buffer));
  EXPECT_EQ(ack_frame.type, FrameType::kAck);

  client.reset();

  auto second = connect_client();
  ASSERT_TRUE(second) << "failed to connect second client";
  second.reset();

  stop.store(true);

  auto drain = connect_client();
  if (drain) {
    drain.reset();
  }

  server_thread.join();

  TelemetryCenter& telemetry = server.telemetry();
  EXPECT_GE(telemetry.GetCounter("ipc_winpipe_accept_total"), 2u);
  EXPECT_EQ(telemetry.GetCounter("ipc_winpipe_handshake_success_total"), 1u);
  EXPECT_EQ(telemetry.GetCounter("ipc_winpipe_handshake_failure_total"), 0u);
  EXPECT_EQ(telemetry.GetCounter("ipc_winpipe_connection_open_total"), 1u);
  EXPECT_EQ(telemetry.GetCounter("ipc_winpipe_disconnect_total"), 1u);
  EXPECT_GT(telemetry.GetCounter("ipc_winpipe_read_bytes_total"), 0u);
  EXPECT_GT(telemetry.GetCounter("ipc_winpipe_write_bytes_total"), 0u);
  EXPECT_EQ(telemetry.GetCounter("ipc_winpipe_accept_error_total"), 0u);
  EXPECT_GE(telemetry.GetCounter("ipc_error_total"),
            static_cast<uint64_t>(error_count.load()));
  EXPECT_DOUBLE_EQ(telemetry.GetGauge("ipc_winpipe_connections_active"), 0.0);
  EXPECT_DOUBLE_EQ(telemetry.GetGauge("ipc_winpipe_pending_ops"), 0.0);
  EXPECT_DOUBLE_EQ(telemetry.GetGauge("ipc_winpipe_write_queue_depth"), 0.0);
}

#endif

}  // namespace
}  // namespace asycipc

